// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package iomisc

import (
	"bytes"
	"io"
	"sync"

	"circuit/kit/rh"
)

// //
// type Prompt interface{}

// //
// type Intr <-chan Prompt

// //
// type Abandon chan<- Prompt

func NewIntr() (rh.Intr, rh.Abandon) {
	ch := make(chan rh.Prompt, 1)
	return rh.Intr(ch), rh.Abandon(ch)
}

//
type InterruptibleReader interface {
	io.ReadCloser
	ReadIntr([]byte, rh.Intr) (int, error)
	Stat() (nrecv int64, closed bool)
}

//
type InterruptibleWriter interface {
	io.WriteCloser
	WriteIntr([]byte, rh.Intr) (int, error)
	Stat() (nsent int64, closed bool)
}

//
func InterruptiblePipe() (InterruptibleReader, InterruptibleWriter) {
	//
	ch := make(chan []byte, 5)
	w := &interruptibleWriter{}
	w.w.ch = ch
	//
	abort := make(chan struct{})
	w.w.abort = abort
	w.a.abort = abort
	//
	r := &interruptibleReader{w: w}
	r.r.ch = ch
	//
	return r, w
}

//
type interruptibleReader struct {
	w *interruptibleWriter
	r struct {
		ch <-chan []byte
		sync.Mutex
		buf bytes.Buffer
	}
	s struct {
		sync.Mutex
		n      int64
		closed bool
	}
}

func (r *interruptibleReader) Read(p []byte) (n int, err error) {
	return r.ReadIntr(p, nil)
}

func (r *interruptibleReader) ReadIntr(p []byte, intr rh.Intr) (n int, err error) {
	r.r.Lock()
	defer r.r.Unlock()
	//
	defer func() {
		r.s.Lock()
		defer r.s.Unlock()
		//
		r.s.n += int64(n)
		if err != nil {
			r.s.closed = true
		}
	}()
	//
	if r.r.buf.Len() > 0 {
		return r.r.buf.Read(p)
	}
	select {
	case block, ok := <-r.r.ch:
		if !ok {
			return 0, io.EOF
		}
		r.r.buf.Write(block)
		return r.r.buf.Read(p)
	case <-intr:
		// Next message is not extracted
		return 0, io.ErrNoProgress
	}
}

func (r *interruptibleReader) Close() error {
	r.w.Close()
	return nil
}

func (r *interruptibleReader) Stat() (nrecv int64, closed bool) {
	r.s.Lock()
	defer r.s.Unlock()
	//
	nrecv, closed = r.s.n, r.s.closed
	return
}

//
type interruptibleWriter struct {
	w struct {
		sync.Mutex
		ch    chan<- []byte
		abort <-chan struct{}
	}
	a struct {
		sync.Mutex
		abort chan<- struct{}
	}
	s struct {
		sync.Mutex
		closed bool
		n      int64
	}
}

func (w *interruptibleWriter) Write(p []byte) (int, error) {
	return w.WriteIntr(p, nil)
}

func (w *interruptibleWriter) WriteIntr(p []byte, intr rh.Intr) (int, error) {
	w.w.Lock()
	defer w.w.Unlock()
	//
	if w.w.ch == nil {
		return 0, io.ErrClosedPipe
	}
	select {
	case w.w.ch <- p:
		w.s.Lock()
		defer w.s.Unlock()
		//
		w.s.n += int64(len(p))
		return len(p), nil
	case <-intr:
		return 0, io.ErrNoProgress
	case <-w.w.abort: // If we receive an abort during write, close out channel here
		close(w.w.ch)
		w.w.ch = nil
		w.close()
		return 0, io.ErrUnexpectedEOF
	}
}

func (w *interruptibleWriter) stop() {
	w.w.Lock()
	defer w.w.Unlock()
	//
	if w.w.ch == nil {
		return
	}
	close(w.w.ch)
	w.w.ch = nil
	//
	w.close()
}

func (w *interruptibleWriter) close() {
	w.s.Lock()
	defer w.s.Unlock()
	//
	w.s.closed = true
}

func (w *interruptibleWriter) abort() {
	w.a.Lock()
	defer w.a.Unlock()
	//
	if w.a.abort == nil {
		return
	}
	close(w.a.abort)
	w.a.abort = nil
}

// Close will interrupt a pending write.
func (w *interruptibleWriter) Close() error {
	w.abort()
	w.stop()
	return nil
}

func (w *interruptibleWriter) Stat() (nsent int64, closed bool) {
	w.s.Lock()
	defer w.s.Unlock()
	//
	return w.s.n, w.s.closed
}

// ReaderEOF returns an io.Reader which will return an io.EOF error as
// soon as the reader r is empty.
func ReaderEOF(r io.Reader) io.Reader {
	return readerEOF{r}
}

type readerEOF struct {
	io.Reader
}

func (x readerEOF) Read(p []byte) (n int, err error) {
	n, err = x.Reader.Read(p)
	if n == 0 && err == nil {
		err = io.EOF
	}
	return
}

// ReadCloserEOF returns an io.ReadCloser which will return an io.EOF error as
// soon as the reader r is empty.
func ReadCloserEOF(r io.ReadCloser) io.ReadCloser {
	return readCloserEOF{r}
}

type readCloserEOF struct {
	io.ReadCloser
}

func (x readCloserEOF) Read(p []byte) (n int, err error) {
	n, err = x.ReadCloser.Read(p)
	if n == 0 && err == nil {
		err = io.EOF
	}
	return
}

// ReaderNopCloser
func ReaderNopCloser(r io.Reader) io.ReadCloser {
	return readerNopCloser{r}
}

type readerNopCloser struct {
	io.Reader
}

func (nr readerNopCloser) Close() error {
	return nil
}

//
func ReaderEOFNopCloser(r io.Reader) io.ReadCloser {
	return ReaderNopCloser(ReaderEOF(r))
}
